package com.caissa.springboot.starter.kafka.consumer;

import com.base.springboot.starter.kafka.consumer.exception.CaissaConsumerDirectStoreErrorMsgException;
import com.base.springboot.starter.kafka.consumer.exception.CaissaConsumerRetryThenStoreErrorMsgException;
import com.base.springboot.starter.kafka.consumer.exception.CaissaIdempotentException;
import com.base.springboot.starter.redis.service.RedisService;
import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.transaction.TransactionStatus;
import org.springframework.transaction.support.TransactionCallback;
import org.springframework.transaction.support.TransactionTemplate;

import java.sql.SQLIntegrityConstraintViolationException;

/**
 * 幂等工具类，定义消息重复、业务异常时消息重试、转存的主逻辑
 * User: wangwei
 * Date: 2021-06-15 3:57 下午
 */
@Slf4j
public abstract class CaissaKafkaMsgIdemHelperDepRedis extends CaissaKafkaMsgIdemHelper{

    @Autowired
    private RedisService redisService;

    @Override
    protected <R> R doExecute(ConsumerRecord consumerRecord, CBack<R> cBack) {
        //// 检查幂等
        checkIdempotent(consumerRecord);

        //// 处理业务
        return cBack.doBiz(consumerRecord);
    }

    /**
     * 幂等检查逻辑，基于redis setnx
     * @param consumerRecord
     * @throws CaissaIdempotentException
     */
    protected void checkIdempotent(ConsumerRecord consumerRecord) throws CaissaIdempotentException {
        try {
            String idepKey = getIdepKey(consumerRecord);
            if (!redisService.setIfAbsent(idepKey, "setIfAbsent", 60)) {
                throw new CaissaIdempotentException("", new RuntimeException("幂等标志 " + idepKey + " 设置失败"));
            }
        } catch (CaissaIdempotentException e) {
            throw e;
        }catch (Exception e) {
            log.error("幂等逻辑执行发生错误，当前没有幂等保护，但不影响主业务逻辑，", e);
        }
    }

    protected abstract String getIdepKey(ConsumerRecord consumerRecord);

}
